#ifndef __M_MESSAGE_H__
#define __M_MESSAGE_H__
#include "../mqcommon/mq_logger.hpp"
#include "../mqcommon/mq_helper.hpp"
#include "../mqcommon/mq_msg.pb.h"


// TODO 为什么先读头四个字节，不是插入的全是字符串吗，没有插入长度吗？
// 那我的思想没有问题，是我的wc格格失误了呀


// valid 和 delivery_mode 的作用完全不同：
// delivery_mode 是 “决定消息是否需要持久化” 的核心标志
// 而 valid 是 “标记已持久化消息是否有效的状态位”，二者没有重叠

namespace zymq
{
    #define DATAFILE_SUBFIX ".mqd"
    #define TMPFILE_SUBFIX ".mqd.tmp"
    using MessagePtr = std::shared_ptr<zymq::Message>;

    class MessageMapper
    {
    public:
        MessageMapper(std::string &parentdir, const std::string &qname)
            :_qname(qname)
        {
            if(parentdir.back() != '/') parentdir += '/';
            _datafile = parentdir + qname + DATAFILE_SUBFIX;
            _tmpfile = parentdir + qname + TMPFILE_SUBFIX;

            assert(FileHelper::createDirectory(parentdir));
            createMsgFile();
        }
        bool createMsgFile()
        {
            if (FileHelper(_datafile).exists() == true) 
                return true;

            bool ret = FileHelper::createFile(_datafile);
            if(ret == false)
            {
                DLOG("创建队列数据文件 %s 失败!", _datafile.c_str());
                return false;
            }
            return true;
        }
        void removeMsgFile()
        {
            FileHelper::removeFile(_datafile);
            FileHelper::removeFile(_tmpfile);
        }
        bool insert(MessagePtr &msg)
        {
            // 函数重载
            return insert(_datafile, msg);
        }
        bool remove(MessagePtr &msg)
        {
            // 消息有效位置为0-》对消息进行序列化-》写入指定位置（即覆盖原有消息的位置）
            msg->mutable_payload()->set_valid("0");
            std::string body = msg->payload().SerializeAsString();
            if(body.size() != msg->length())
            {
                DLOG("不能修改文件中的数据信息，因为新生成的数据与原数据长度不一致!");
                return false; 
            }
            FileHelper helper(_datafile);
            bool ret = helper.write(body.c_str(), msg->offset(), msg->length());
            if(ret == false)
            {
                DLOG("向队列数据文件 %s 写入数据失败!",_datafile.c_str());
                return false;
            }
            return true;
        }
        std::list<MessagePtr> gc()
        {
            std::list<MessagePtr> result;
            bool ret;

            // 1. 先加载文件中的有效消息 
            // 4字节长度|有效载荷|4字节长度|有效载荷|4字节长度|有效载荷.......
            ret = load(result);
            if(ret == false)
            {
                DLOG("加载有效消息失败!")
                return result;
            }
            // DLOG("垃圾回收，得到有效消息数量: %ld", result.size());

            // 2. 将有效数据，进行序列化存储到临时文件中
            // 注意这里必须保证临时文件存在
            // 比如：在重启时且queue.mqd文件中没有内容，此时范围for直接出来
            // 直接进去删原文件，而此时没有创建临时文件，就导致了删除了原始的文件名称，此时再更改临时文件名，则报错
            // 而我们在此处创建了临时文件，就能保证rename一定成功，保证原文件名一定存在
            FileHelper::createFile(_tmpfile);
            
            for(auto & msg : result)
            {
                ret = insert(_tmpfile, msg);
                if(ret == false)
                {
                    DLOG("向临时文件写入消息数据失败!");
                    return result;
                }
            }
            // DLOG("向临时文件写入完毕，临时文件大小：%ld", FileHelper(_tmpfile).size());

            // 3. 删除原文件
            ret = FileHelper::removeFile(_datafile);
            if(ret == false)
            {
                DLOG("删除原文件失败!");
                return result;
            }

            // 4. 修改临时文件名为原文件名称
            ret = FileHelper(_tmpfile).rename(_datafile);
            if(ret == false)
            {
                DLOG("修改临时文件名为原文件名称失败!");
                return result;
            }

            // 5. 返回新的有效数据
            return result;
        }
    private:
        // 从datafile文件中读取所有的有效消息，放入gc函数的消息链表引用中
        bool load(std::list<MessagePtr> &result)
        {
            bool ret;

            FileHelper helper(_datafile);
            size_t offset = 0, msg_size, fsize = helper.size();
            while(offset < fsize)
            {
                ret = helper.read((char*)&msg_size, offset, sizeof(size_t));
                if(ret == false)
                {
                    DLOG("读取消息长度失败!");
                    return false;
                }
                offset += sizeof(size_t);
                
                std::string msg_body(msg_size, '\0');
                ret = helper.read(&msg_body[0], offset, msg_size);
                if(ret == false)
                {
                    DLOG("读取有效载荷失败!");
                    return false;
                }
                offset += msg_size;

                MessagePtr msgp = std::make_shared<Message>();
                msgp->mutable_payload()->ParseFromString(msg_body);
                // 如果是无效消息，处理下一个
                if(msgp->payload().valid() == "0") continue;
                // 有效消息则保存起来
                result.push_back(msgp);
            }

            return true;
        }
        bool insert(const std::string &filename, MessagePtr &msg)
        {
            // 序列化消息-》获取文件长度-》写入指定位置-》更新msg中的实际存储信息
            std::string body = msg->payload().SerializeAsString();
            
            FileHelper helper(filename);

            // 获取应该写入的位置,即文件的末尾，写四字节长度
            size_t fsize = helper.size();
            size_t msg_head = body.size();
            bool ret = helper.write((char*)&msg_head, fsize, sizeof(size_t));
            if(ret == false)
            {
                DLOG("向队列数据文件 %s 写入消息头长度失败!",_datafile.c_str());
                return false;
            }

            // 获取应该写入的位置,即文件的末尾，写消息体
            fsize = helper.size();
            ret = helper.write(body.c_str(), fsize, body.size());
            if(ret == false)
            {
                DLOG("向队列数据文件 %s 写入有效载荷失败!",_datafile.c_str());
                return false;
            }
            msg->set_offset(fsize);
            msg->set_length(body.size());
            return true;
        }
    private:
        std::string _qname;
        std::string _datafile;
        std::string _tmpfile;
    };

    class QueueMessage
    {
    public:
        using ptr = std::shared_ptr<QueueMessage>;

        QueueMessage(std::string &parentdir, const std::string &qname)
            :_qname(qname), _valid_count(0), _total_count(0), _mapper(parentdir, qname){}

        void recovery()
        {
            // gc就是回收文件中的垃圾数据，同时返回一个新消息list
            std::unique_lock<std::mutex> lock(_mutex);
            
            _msgs = _mapper.gc();

            for(auto & msg : _msgs)
                _durable_msgs.insert({msg->payload().properties().id(), msg});
            _valid_count = _total_count = _msgs.size();
        }
        // 这个delivery_mode貌似是队列的是否持久化？？TODO
        bool insert(const BasicProperties *bp, const std::string &body, bool queue_is_durable)
        {
            // 1. 构造消息对象
            MessagePtr msg = std::make_shared<Message>();
            msg->mutable_payload()->set_body(body);
            if(bp != nullptr)
            {
                // 该消息的属性不为空，该消息的持久化状态取决于 队列的持久状态和消息的持久化状态
                DeliveryMode mode = queue_is_durable ? bp->delivery_mode() : DeliveryMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_id(bp->id());
                msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
            }
            else
            {
                DeliveryMode mode = queue_is_durable ? DeliveryMode::DURABLE : DeliveryMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_id(UUIDHelper::uuid());
                msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key("");
            }

            std::unique_lock<std::mutex> lock(_mutex);

            // 2. 判断是否需要持久化存储
            if(msg->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                // 表示在文件存储中的这条数据有效
                msg->mutable_payload()->set_valid("1");

                // 3. 进行持久化存储
                bool ret = _mapper.insert(msg);
                if(ret == false)
                {
                    DLOG("持久化存储信息: %s 失败了!", body.c_str());
                    return false;
                }
                _valid_count += 1; // 文件中的有效消息
                _total_count += 1;
                _durable_msgs.insert({msg->payload().properties().id(), msg});
            }

            // 4. 内存的管理
            _msgs.push_back(msg);
            return true;
        }
        MessagePtr front()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            
            if(_msgs.size() == 0) return MessagePtr();

            MessagePtr msg = _msgs.front();
            _msgs.pop_front();

            _waitack_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            return msg;
        }
        bool remove(const std::string &msg_id)
        {
            std::unique_lock<std::mutex> lock(_mutex);

            // 1. 从待确认队列中查找消息
            auto it = _waitack_msgs.find(msg_id);
            if(it == _waitack_msgs.end())
            {
                DLOG("没有找到要删除的信息： %s!", msg_id.c_str());
                return true;
            }

            // 2. 根据消息的持久化模式，决定是否删除持久化信息
            if(it->second->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                // 3. 删除持久化信息
                _mapper.remove(it->second);
                _durable_msgs.erase(msg_id);
                _valid_count -= 1;
                QueueMessage::gc(); // 内部判断是否需要垃圾回收
            }

            // 4. 删除内存中的信息
            _waitack_msgs.erase(msg_id);
            return true;
        }
        
        // 消息队列中待发送的数量
        size_t getable_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _msgs.size();
        }
        size_t total_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _total_count;
        }
        size_t durable_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _durable_msgs.size();
        }
        size_t waitack_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _waitack_msgs.size();
        }
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _mapper.removeMsgFile();
            _msgs.clear();
            _durable_msgs.clear();
            _waitack_msgs.clear();
            _valid_count = _total_count = 0;
        }
    private:
        bool GCCheck()
        {
            // 总消息数量大于2000 并且有效消息不足50%
            if(_total_count > 2000 && _valid_count * 10 / _total_count < 5)
                return true;
            return false; 
        }
        void gc()
        {
            // 1. 进行垃圾回收，获取到垃圾回收后，有效的消息信息链表
            if(GCCheck() == false) return;
            std::list<MessagePtr> msgs = _mapper.gc();
            for(auto & msg : msgs)
            {
                auto it = _durable_msgs.find(msg->payload().properties().id());
                if(it == _durable_msgs.end())
                {
                    DLOG("垃圾回收后，有一条持久化消息，在内存中没有进行管理!");
                    _msgs.push_back(msg);// 重新添加到推送链表末尾
                    _durable_msgs.insert({msg->payload().properties().id(), msg});
                    continue;
                }
                // 更新每一条消息的实际存储位置
                it->second->set_offset(msg->offset());
                it->second->set_length(msg->length());
            }
            // 更新当前的有效消息数量 & 总的持久化消息数量
            _total_count = _valid_count = msgs.size();
        }
    private:
        std::mutex _mutex;
        std::string _qname;
        size_t _valid_count; // 在文件中的有效消息数量
        size_t _total_count; // 在文件中存储的总的消息数量
        MessageMapper _mapper;
        std::list<MessagePtr> _msgs; // 待推送消息 可能会有持久化消息和非持久化消息
        std::unordered_map<std::string, MessagePtr> _durable_msgs; // 持久化消息hash
        std::unordered_map<std::string, MessagePtr> _waitack_msgs; // 待确认消息hash
    };

    class MessageManager
    {
    public:
        using ptr = std::shared_ptr<MessageManager>;

        MessageManager(const std::string &parentdir):_parentdir(parentdir) {}

        void initQueueMessage(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it != _queue_msgs.end())
                {
                    return;
                }
                qmp = std::make_shared<QueueMessage>(_parentdir, qname);
                _queue_msgs.insert({qname, qmp});
            }
            qmp->recovery();
        }
        void destroyQueueMessage(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    return;
                }
                qmp = it->second;
                _queue_msgs.erase(it);
            }

            qmp->clear();
        }
        bool insert(const std::string &qname, BasicProperties *bp, const std::string &body, bool queue_is_durable)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    DLOG("向队列%s新增消息失败: 没有找到消息管理句柄!", qname.c_str());
                    return false;
                }
                qmp = it->second;
            }
            return qmp->insert(bp, body, queue_is_durable);
        }
        MessagePtr front(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    DLOG("获取队列%s队首消息失败: 没有找到消息管理句柄!", qname.c_str());
                    return MessagePtr();
                }
                qmp = it->second;
            }
            return qmp->front();
        }
        void ack(const std::string &qname, const std::string &msg_id)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    DLOG("确认队列%s消息%s失败: 没有找到消息管理句柄!", qname.c_str(), msg_id.c_str());
                    return ;
                }
                qmp = it->second;
            }
            qmp->remove(msg_id);
        }
        size_t getable_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    DLOG("获取队列%s待推送消息数量失败: 没有找到消息管理句柄!", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->getable_count();
        }
        size_t total_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    DLOG("获取队列%s总持久化消息数量失败: 没有找到消息管理句柄!", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->total_count();
        }
        size_t durable_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    DLOG("获取队列%s有效持久化消息数量失败: 没有找到消息管理句柄!", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->durable_count();
        }
        size_t waitack_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    DLOG("获取队列%s待确认消息数量失败: 没有找到消息管理句柄!", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->waitack_count();
        }
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            for(auto & qmsgs : _queue_msgs)
                qmsgs.second->clear();
        }
    private:
        std::mutex _mutex;
        std::string _parentdir;
        std::unordered_map<std::string, QueueMessage::ptr> _queue_msgs;
    };
}

#endif